package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.OrderDetail;
import com.atguigu.gmall.realtime.bean.OrderInfo;
import com.atguigu.gmall.realtime.bean.OrderWide;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author: xu
 * @desc: 合并订单宽表
 * 程  序：MockDB -> MySQL -> FlinkCDC -> Kafka(ZK) -> BaseDbApp -> Kafka/Phoenix(zk/hdfs/hbase) -> OrderWideApp(Redis) -> Kafka
 * 测试需要启动服务：
 * - Maxwell、zk、kafka、hdfs、hbase、BaseDBApp OrderWideApp 准备配置表
 * 业务执行流程：
 * - 1.模拟生成数据jar
 * - 2.在MySQL数据库的表中插入相关的业务数据
 * - 3.MySQL的Binlog记录变化的数据
 * - 4.Maxwell会将变化的数据采集到，并且发送到Kafka的ods_base_db_m
 * - 5.BaseDBApp从ods_base_db_m主题中读取数据，进行分流操作
 * - > 事实：写回到kafka的dwd主题
 * - > 维度：保存到phoenix的维度表中
 * - 6.OrderWideApp从kafka的dwd层和HBase的维度表中读取数据
 * - 7.使用的是intervalJoin对OrderInfo和OrderDetail进行join
 * - 8.将用户维度关联到订单宽表上
 * - > 基本的维度关联
 * - > 优化1：旁路缓存的优化：先从Redis中查询维度，命中直接获取，否则再到数据库查询，并将查询结果放到Redis中，如果维度数据发生变化，清除缓存
 * - > 优化2：异步IO：自定义类继承RichAsyncFunction，重写asyncInvoke，从线程池中获取新的线程，并执行维度关联操作，使用模板方法设计模式
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/OrderWideApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");

        // TODO 2.从Kafka的DWD层读取订单和订单明细数据
        // 2.1 声明相关的主题以及消费者组
        String orderInfoSourceTopic = "dwd_order_info";
        String orderDetailSourceTopic = "dwd_order_detail";
        String groupId = "order_wide_group";
        // 2.2 读取订单主题数据
        DataStreamSource<String> orderInfoStrStream = env.addSource(MyKafkaUtil.getKafkaSource(orderInfoSourceTopic, groupId));
        // 2.3 读取订单明细数据
        DataStreamSource<String> orderDetailStrStream = env.addSource(MyKafkaUtil.getKafkaSource(orderDetailSourceTopic, groupId));

        // TODO 3.对读取的数据进行结构的转换，jsonString -->OrderInfo|OrderDetail
        // 3.1 转换订单数据结构
        SingleOutputStreamOperator<OrderInfo> orderInfoStream = orderInfoStrStream.map(
                new RichMapFunction<String, OrderInfo>() {
                    private FastDateFormat fdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        fdf = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderInfo map(String jsonStr) throws Exception {
                        OrderInfo orderInfo = JSON.parseObject(jsonStr, OrderInfo.class);
                        orderInfo.setCreate_ts(fdf.parse(orderInfo.getCreate_time()).getTime());
                        return orderInfo;
                    }
                }
        );

        // 3.2 转换订单明细数据结构
        SingleOutputStreamOperator<OrderDetail> orderDetailStream = orderDetailStrStream.map(
                new RichMapFunction<String, OrderDetail>() {
                    private FastDateFormat fdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        fdf = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderDetail map(String jsonStr) throws Exception {
                        OrderDetail orderDetail = JSON.parseObject(jsonStr, OrderDetail.class);
                        orderDetail.setCreate_ts(fdf.parse(orderDetail.getCreate_time()).getTime());
                        return orderDetail;
                    }
                }
        );
        // orderInfoStream.print("orderInfo>>>");
        // orderDetailStream.print("orderDetail>>>");

        // TODO 4.设定事件时间水位，然后按照订单id进行分组，指定关联的key
        // 4.1 订单指定事件时间字段，按照ID分组
        KeyedStream<OrderInfo, Long> orderInfoKeyedStream = orderInfoStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((orderInfo, recordTimestamp) -> orderInfo.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);
        // 4.2 订单明细指定事件时间字段，，按照ID分组
        KeyedStream<OrderDetail, Long> orderDetailKeyedStream = orderDetailStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((orderDetail, recordTimestamp) -> orderDetail.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);


        // TODO 5.使用intervalJoin对订单和订单明细进行关联
        SingleOutputStreamOperator<OrderWide> orderWideSteam = orderInfoKeyedStream
                .intervalJoin(orderDetailKeyedStream)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                             @Override
                             public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context ctx, Collector<OrderWide> out) throws Exception {
                                 out.collect(new OrderWide(orderInfo, orderDetail));
                             }
                         }
                );
        // orderWideSteam.print("orderWide>>>>");

        // TODO 6.关联用户维度，HBase Phoenix
        SingleOutputStreamOperator<OrderWide> orderWideWithUserStream = AsyncDataStream.unorderedWait(
                orderWideSteam,
                new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getUser_id().toString();
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                        // 获取用户生日
                        String birthday = dimInfoJsonObj.getString("BIRTHDAY");
                        // 定义日期转换工具类
                        FastDateFormat fdf = FastDateFormat.getInstance("yyyy-MM-dd");
                        // 获取生日日期的毫秒数
                        long birthdayTs = fdf.parse(birthday).getTime();
                        // 年龄毫秒数
                        long ageTs = System.currentTimeMillis() - birthdayTs;
                        // 转换为年龄
                        long ageLong = ageTs / 1000L / 60L / 60L / 24L / 365L;
                        // 将维度中的年龄赋值给订单宽表中的属性
                        orderWide.setUser_age((int) ageLong);
                        // 将维度中的性别赋值给订单宽表中的属性
                        orderWide.setUser_gender(dimInfoJsonObj.getString("GENDER"));
                    }
                },
                60, TimeUnit.SECONDS);
        // orderWideWithUserStream.print();

        // TODO 7.关联省市维度
        SingleOutputStreamOperator<OrderWide> orderWideWithProvinceStream = AsyncDataStream.unorderedWait(
                orderWideWithUserStream,
                new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getProvince_id().toString();
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                        orderWide.setProvince_name(dimInfoJsonObj.getString("NAME"));
                        orderWide.setProvince_area_code(dimInfoJsonObj.getString("AREA_CODE"));
                        orderWide.setProvince_iso_code(dimInfoJsonObj.getString("ISO_CODE"));
                        orderWide.setProvince_3166_2_code(dimInfoJsonObj.getString("ISO_3166_2"));
                    }
                },
                60, TimeUnit.SECONDS
        );
        // orderWideWithProvinceStream.print(">>>>>");

        // TODO 8.关联SKU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSkuStream = AsyncDataStream.unorderedWait(
                orderWideWithProvinceStream,
                new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSku_id());
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                        orderWide.setSku_name(dimInfoJsonObj.getString("SKU_NAME"));
                        orderWide.setSpu_id(dimInfoJsonObj.getLong("SPU_ID"));
                        orderWide.setCategory3_id(dimInfoJsonObj.getLong("CATEGORY3_ID"));
                        orderWide.setTm_id(dimInfoJsonObj.getLong("TM_ID"));
                    }
                }, 60, TimeUnit.SECONDS
        );
        // orderWideWithSkuStream.print(">>>>");

        // TODO 9.关联SPU商品维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuStream = AsyncDataStream.unorderedWait(
                orderWideWithSkuStream,
                new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSpu_id());
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfoJsonObj) throws Exception {
                        orderWide.setSpu_name(dimInfoJsonObj.getString("SPU_NAME"));
                    }
                }, 60, TimeUnit.SECONDS
        );
        // orderWideWithSpuStream.print(">>>>>");

        // TODO 10.关联品类维度
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3Stream = AsyncDataStream.unorderedWait(
                orderWideWithSpuStream,
                new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setCategory3_name(jsonObject.getString("NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getCategory3_id());
                    }
                }, 60, TimeUnit.SECONDS);
        // orderWideWithCategory3Stream.print(">>>>>");

        // TODO 11.关联品牌维度
        SingleOutputStreamOperator<OrderWide> orderWideWithTmStream = AsyncDataStream.unorderedWait(
                orderWideWithCategory3Stream,
                new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getTm_id());
                    }
                }, 60, TimeUnit.SECONDS);
        orderWideWithTmStream.print(">>>>>");

        // TODO 12.将关联后的订单宽表数据写回到kafka的DWM层
        String orderWideSinkTopic = "dwm_order_wide";
        orderWideWithTmStream
                .map(JSON::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink(orderWideSinkTopic));

        env.execute(OrderWideApp.class.getSimpleName());
    }
}
